package check_message_back

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/gin-gonic/gin"
	"github.com/segmentio/kafka-go"
	"go-demo-2025/common"
	"go-demo-2025/utils/kafkaUtil"
	"log"
	"net/http"
	"strings"
	"time"
)

// kafka 生产者: 消息回查机制方案：Kafka 原生不支持事务消息，可通过 “准备 - 提交” 两阶段模式实现消息回查机制，确保业务成功后才投递消息。

// 常量配置
const (
	lookbackTopic = "lookback_topic" // 回查主题（传递准备/提交/回滚消息）
)

// LookbackMessage 回查消息通用结构体
type LookbackMessage struct {
	MsgType     string `json:"msg_type"`     // "prepare"/"commit"/"rollback"
	BizType     string `json:"biz_type"`     // 业务类型（如"order_create"）
	BizID       string `json:"biz_id"`       // 业务唯一ID（如订单号）
	TargetTopic string `json:"target_topic"` // 目标业务主题
	Content     string `json:"content"`      // 消息内容
	CallbackURL string `json:"callback_url"` // 业务回查接口
	OperateTime int64  `json:"operate_time"` // 操作时间戳（毫秒）
}

// 全局存储未确认的准备消息（生产环境建议用Redis/数据库替代）
var unconfirmedMsgs = make(map[string]LookbackMessage) // key: bizType:bizID

// 初始化回查服务依赖（创建回查主题）
func InitLookbackService() error {
	// 检查并创建回查主题
	exists, err := kafkaUtil.CheckTopicExists(lookbackTopic)
	if err != nil {
		return fmt.Errorf("检查回查主题失败: %v", err)
	}
	if !exists {
		if err := kafkaUtil.CreateTopic(lookbackTopic); err != nil {
			return fmt.Errorf("创建回查主题失败: %v", err)
		}
		time.Sleep(2 * time.Second)
		log.Println("回查主题初始化成功")
	}
	return nil
}

// 1. 发送准备消息（第一阶段：业务执行前）
func SendPrepareMessage(ctx *gin.Context, bizType, bizID, targetTopic, content string) error {
	msg := LookbackMessage{
		MsgType:     "prepare",
		BizType:     bizType,
		BizID:       bizID,
		TargetTopic: targetTopic,
		Content:     content,
		CallbackURL: fmt.Sprintf("http://localhost:%s/api/kafka/callback", ctx.Request.Host),
		OperateTime: time.Now().UnixMilli(),
	}

	msgBytes, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("序列化准备消息失败: %v", err)
	}

	// 发送到回查主题
	return kafkaUtil.SendKafkaMessage(ctx, lookbackTopic, fmt.Sprintf("%s:%s", bizType, bizID), string(msgBytes))
}

// 2. 发送确认消息（第二阶段：业务执行后提交/回滚）
// 新增参数：targetTopic 和 content（提交时需要，回滚时可传空）
func SendConfirmMessage(ctx *gin.Context, bizType, bizID, confirmType, targetTopic, content string) error {
	msg := LookbackMessage{
		MsgType:     confirmType, // "commit" 或 "rollback"
		BizType:     bizType,
		BizID:       bizID,
		TargetTopic: targetTopic, // 提交时携带目标主题
		Content:     content,     // 提交时携带消息内容
		OperateTime: time.Now().UnixMilli(),
	}

	msgBytes, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("序列化确认消息失败: %v", err)
	}

	// 发送到回查主题
	return kafkaUtil.SendKafkaMessage(ctx, lookbackTopic, fmt.Sprintf("%s:%s", bizType, bizID), string(msgBytes))
}

// 3. 业务接口：创建订单（集成回查机制）
func CreateOrderHandler(ctx *gin.Context) {
	var req struct {
		OrderID string `json:"order_id" binding:"required"`
		Amount  int64  `json:"amount" binding:"required,min=1"`
	}
	if err := ctx.ShouldBindJSON(&req); err != nil {
		ctx.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	bizType := "order_create"
	bizID := req.OrderID
	targetTopic := "order_events" // 最终业务主题
	content := fmt.Sprintf(`{"order_id":"%s","amount":%d,"status":"pending"}`, req.OrderID, req.Amount)

	// 步骤1：发送准备消息
	if err := SendPrepareMessage(ctx, bizType, bizID, targetTopic, content); err != nil {
		ctx.JSON(http.StatusInternalServerError, gin.H{"error": "发送准备消息失败: " + err.Error()})
		return
	}

	// 步骤2：执行本地业务（创建订单）
	tx, err := common.SqlDB.BeginTx(ctx, nil)
	if err != nil {
		ctx.JSON(http.StatusInternalServerError, gin.H{"error": "开启事务失败: " + err.Error()})
		return
	}
	defer tx.Rollback()

	// 创建订单记录
	_, err = tx.ExecContext(ctx, `
		INSERT INTO orders (order_id, amount, status, create_time) 
		VALUES (?, ?, 'pending', NOW())
	`, req.OrderID, req.Amount)
	if err != nil {
		ctx.JSON(http.StatusInternalServerError, gin.H{"error": "创建订单失败: " + err.Error()})
		return
	}

	// 步骤3：业务成功，发送提交消息（补充 targetTopic 和 content）
	if err := SendConfirmMessage(ctx, bizType, bizID, "commit", targetTopic, content); err != nil {
		log.Printf("发送提交消息失败（回查服务将兜底）: %v", err)
	}

	// 提交数据库事务
	if err := tx.Commit(); err != nil {
		ctx.JSON(http.StatusInternalServerError, gin.H{"error": "提交事务失败: " + err.Error()})
		return
	}

	ctx.JSON(http.StatusOK, gin.H{"success": true, "order_id": req.OrderID})
}

// 4. 业务回查接口（供回查服务调用）
func CallbackHandler(ctx *gin.Context) {
	var req struct {
		BizType string `json:"biz_type" binding:"required"`
		BizID   string `json:"biz_id" binding:"required"`
	}
	if err := ctx.ShouldBindJSON(&req); err != nil {
		ctx.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	// 查询订单状态（确认业务是否成功）
	var status string
	err := common.SqlDB.QueryRowContext(ctx, `
		SELECT status FROM orders WHERE order_id = ?
	`, req.BizID).Scan(&status)
	if err != nil {
		// 订单不存在 → 业务失败
		ctx.JSON(http.StatusOK, gin.H{"status": "rollback"})
		return
	}

	// 根据订单状态返回结果
	switch status {
	case "pending", "paid", "success":
		ctx.JSON(http.StatusOK, gin.H{"status": "commit"})
	case "cancelled", "failed":
		ctx.JSON(http.StatusOK, gin.H{"status": "rollback"})
	default:
		ctx.JSON(http.StatusOK, gin.H{"status": "unknown"})
	}
}

// 5. 回查服务核心（消费回查主题+定时兜底）
func StartLookbackService(ctx context.Context) {
	// 初始化Kafka消费者（监听回查主题）
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  kafkaUtil.KafkaBrokers,
		Topic:    lookbackTopic,
		GroupID:  "lookback_consumer_group",
		MaxWait:  5 * time.Second,
		MinBytes: 1024,
		MaxBytes: 1024 * 1024,
	})
	defer reader.Close()

	// 定时任务：每30秒扫描未确认消息
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	log.Println("回查服务启动成功")

	for {
		select {
		case <-ctx.Done():
			log.Println("回查服务退出")
			return

		// 定时兜底：处理超时未确认的消息
		case <-ticker.C:
			handleTimeoutMessages()

		// 实时消费回查主题消息
		default:
			msg, err := reader.ReadMessage(ctx)
			if err != nil {
				log.Printf("消费回查消息失败: %v", err)
				continue
			}

			// 解析消息
			var lbMsg LookbackMessage
			if err := json.Unmarshal(msg.Value, &lbMsg); err != nil {
				log.Printf("解析回查消息失败: %v, 原始数据: %s", err, string(msg.Value))
				continue
			}

			key := fmt.Sprintf("%s:%s", lbMsg.BizType, lbMsg.BizID)
			switch lbMsg.MsgType {
			case "prepare":
				// 新增未确认消息
				unconfirmedMsgs[key] = lbMsg
				log.Printf("收到准备消息: %s", key)

			case "commit":
				// 提交 → 投递业务消息+移除未确认记录
				handleCommitMessage(lbMsg, key)

			case "rollback":
				// 回滚 → 直接移除未确认记录
				delete(unconfirmedMsgs, key)
				log.Printf("处理回滚消息: %s", key)
			}
		}
	}
}

// 处理提交消息：投递到业务主题
func handleCommitMessage(msg LookbackMessage, key string) {
	if _, exists := unconfirmedMsgs[key]; !exists {
		log.Printf("提交消息对应准备消息不存在: %s", key)
		return
	}

	// 投递消息到目标业务主题
	err := kafkaUtil.SendKafkaMessage(context.Background(), msg.TargetTopic, key, msg.Content)
	if err != nil {
		log.Printf("投递业务消息失败: %v, 消息: %+v", err, msg)
		return
	}

	// 处理成功，移除未确认记录
	delete(unconfirmedMsgs, key)
	log.Printf("提交消息处理成功: %s", key)
}

// 处理超时未确认的准备消息
func handleTimeoutMessages() {
	timeoutThreshold := 5 * 60 * 1000 // 5分钟超时
	now := time.Now().UnixMilli()

	for key, msg := range unconfirmedMsgs {
		// 只处理超时消息
		if now-msg.OperateTime < int64(timeoutThreshold) {
			continue
		}

		// 调用业务回查接口确认状态
		confirmStatus, err := callCallbackAPI(msg.CallbackURL, msg.BizType, msg.BizID)
		if err != nil {
			log.Printf("回查接口调用失败: %v, 消息: %+v", err, msg)
			continue
		}

		switch confirmStatus {
		case "commit":
			// 确认提交 → 投递消息
			handleCommitMessage(msg, key)
		case "rollback":
			// 确认回滚 → 移除记录
			delete(unconfirmedMsgs, key)
			log.Printf("超时消息回滚: %s", key)
		case "unknown":
			// 未知状态 → 保留，下次重试
			log.Printf("超时消息状态未知，下次重试: %s", key)
		}
	}
}

// 调用业务回查接口
func callCallbackAPI(url, bizType, bizID string) (string, error) {
	// 构造请求体
	reqBody, err := json.Marshal(map[string]string{
		"biz_type": bizType,
		"biz_id":   bizID,
	})
	if err != nil {
		return "", fmt.Errorf("构造请求体失败: %v", err)
	}

	// 发送POST请求
	resp, err := http.Post(url, "application/json", strings.NewReader(string(reqBody)))
	if err != nil {
		return "", fmt.Errorf("请求回查接口失败: %v", err)
	}
	defer resp.Body.Close()

	// 解析响应
	var result map[string]string
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return "", fmt.Errorf("解析回查响应失败: %v", err)
	}

	return result["status"], nil
}
